package com.sxhuayuan.parking.queue.rabbitmq.consumer;

import java.io.IOException;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.sxhuayuan.parking.compenent.RabbitMQConfig;
import com.sxhuayuan.parking.queue.msg.QueueMsg;
import com.sxhuayuan.parking.queue.rabbitmq.RabbitMQ;

/**
 * 延迟消息变为死信消息后的处理
 * 
 * @author gy
 */
@Component("delayDeadLetterMsgCustomer")
@RabbitListener(queues = RabbitMQConfig.QUEUE_DELAY_DLX)
public class DelayDeadLetterMsgCustomer {

	Logger log = LoggerFactory.getLogger(getClass());

	@Autowired
	RabbitTemplate rabbitTmpl;

	@RabbitHandler()
	public void customer(@Payload byte[] bytes, Channel channel, @Headers Map<String, Object> headers) {
		try {
			String str = new String(bytes);
			log.debug("延时死信: {}", str);
			log.debug("headers: {}", headers);
			long tag = Long.valueOf(headers.get(AmqpHeaders.DELIVERY_TAG).toString());
			String routingKey = headers.get(RabbitMQ.X_FIRST_DEAD_QUEUE).toString();
			QueueMsg msg = JSON.parseObject(str, QueueMsg.class);
			if (msg.isDeadMsg()) {
				log.debug("[{}]死信超过重试次数", msg.getId());
				// 超过最大重试次数，进入错误队列，警报处理
				channel.basicNack(tag, false, false);
			} else {
				MessageProperties properties = MessagePropertiesBuilder.newInstance().build();
				Integer ttr = msg.nextTtr();
				String json = JSON.toJSONString(msg);
				log.debug("[{}]死信重新延时: {}s", msg.getId(), ttr);
				// 更新延迟时间，放入队列
				Message reMsg = MessageBuilder.withBody(json.getBytes()).andProperties(properties).setHeader("x-delay", ttr * 1000).build();
				rabbitTmpl.send(RabbitMQConfig.EXCHANGE_DELAY_START, routingKey, reMsg);
				channel.basicAck(tag, false);
			}
		} catch (IOException e) {
			log.error("", e);
		}
	}
}
